package org.ring.config;

import org.ring.common.URL;
import org.ring.common.transport.Connection;
import org.ring.common.transport.ConnectionListener;
import org.ring.common.transport.MessageHandler;
import org.ring.common.transport.DefaultMessageHandlerRegistry;
import org.ring.common.transport.TcpServerTransport;
import org.ring.config.common.GetRequest;
import org.ring.config.common.GetResponse;
import org.ring.config.server.SubscribeRelation;
import org.ring.config.server.Subscriber;

import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @since 2023-10-28 11:39
 */
public class ServerConfigTransportTest {



    public static void main(String[] args) throws Exception{


        final SubscribeRelation<String> subscribeRelation = new SubscribeRelation();
        URL url = new URL("transport","",12300);
        TcpServerTransport serverTransport = new TcpServerTransport(url);
        serverTransport.getConnectionManager().addConnectionListener(new ConnectionListener() {
            @Override
            public void onDisConnected(Connection connection) {
                subscribeRelation.remove(connection);
            }
        });

        DefaultMessageHandlerRegistry messageHandlerRegistry = new DefaultMessageHandlerRegistry();
        messageHandlerRegistry.register(GetRequest.class,
            (MessageHandler<GetRequest, GetResponse>) (connection, input) -> {
                String result = "server recevied ::::" + input;

                Subscriber subscriber = new Subscriber(connection,input.getConfigId(),System.currentTimeMillis() + "");
                subscribeRelation.add(input.getConfigId(),subscriber);

                GetResponse getResponse = new GetResponse();
                getResponse.setVersion(System.currentTimeMillis() + "");
                getResponse.setContent(result);
                return CompletableFuture.completedFuture(getResponse);
            });


        serverTransport.handlerRegistry(messageHandlerRegistry);
        serverTransport.start();

        final AtomicLong version = new AtomicLong();
        Thread t = new Thread(()-> {
            while (true){
                Collection<Connection> subscribers = subscribeRelation.getSubscribers("test");
                if(subscribers != null){
                    subscribers.forEach(conn -> {
                        if(conn != null){
                            GetResponse getResponse = new GetResponse();
                            getResponse.setVersion(version.getAndIncrement() + "");
                            getResponse.setContent("test");
                            CompletableFuture future = conn.request(getResponse,3000,null);
                            future.whenComplete((r, e) -> {
                                System.out.println("response---> " + r);
                            });
                        }
                    });
                }


                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        t.start();
    }
}
